package com.ls.sort;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.TreeSet;

/**
 * 文件排序
 *
 * @author : 梁苏
 * @date : 2023/3/17 16:18
 **/
public class FileSort<T> {

    private List<FileData<T>> fileDatas = new ArrayList<>();

    private List<T> list = new ArrayList<>();

    private Comparator<T> comparator;

    // 多少条数据写一次文件
    private int writeToFileNum = 1000;

    private ObjectWriter mergeObjectWriter;

    public FileSort(Comparator<T> comparator, int writeToFileNum){
        this.comparator = comparator;
        this.writeToFileNum = writeToFileNum;
    }

    public void add(T obj) {
        list.add(obj);
        if (list.size() == writeToFileNum) {
            list.sort(comparator);
            writeToFile(list);
            list.clear();
        }
    }

    private void writeToFile(List<T> list) {
        String fileName = "temp_%d.data";
        File file = new File(String.format(fileName, fileDatas.size()));

        FileData fileData = new FileData(file, list.size(), list.get(0).getClass(), new HeapByteBuf(1024));
        fileData.writeToFile(list);

        fileDatas.add(fileData);
    }

    private void writeToMergeFile(List<T> list) {
        if (mergeObjectWriter == null) {
            mergeObjectWriter = new ObjectWriter(
                    list.get(0).getClass(), new HeapByteBuf(1024), new File("merge.data"));
        }
        try {
            for (T t : list) {
                mergeObjectWriter.writeObject(t);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void sort() {
        int fetchSize = writeToFileNum / fileDatas.size();

        // 每次拿取数量
        for (FileData<T> fileData : fileDatas) {
            fileData.setFetchSize(fetchSize);
        }

        List<T> list = new ArrayList<>(writeToFileNum);

        // 有数据的文件
        TreeSet<FileData<T>> treeSet = new TreeSet<>((o1, o2) -> comparator.compare(o1.getNext(), o2.getNext()));
        treeSet.addAll(fileDatas);

        while (!treeSet.isEmpty()) {
            // 拿出最小值
            FileData<T> fileData = treeSet.pollFirst();
            // 放入最小值
            T next = fileData.next();
            list.add(next);
            System.out.println("排序后：" + next);

            // 写文件
            if (list.size() == writeToFileNum) {
                writeToMergeFile(list);
                list.clear();
            }

            // 还有数据
            if (fileData.hasNext()) {
                treeSet.add(fileData);
            }
        }
    }

}

class FileData<T> {

    private ObjectWriter<T> writer;

    private ObjectReader<T> reader;

    private List<T> partList = new ArrayList<>();

    private Integer fetchStart = 0;

    private Integer fetchEnd = 0;

    private Integer readIndex = 0;

    private Integer writeIndex = 0;

    private Integer fetchSize;

    public FileData(File file, Integer dataNum, Class<T> clazz, ByteBuf buf){
        this.writeIndex = dataNum;
        this.reader = new ObjectReader(clazz, buf, file);
        this.writer = new ObjectWriter<>(clazz, buf, file);
    }

    public void setFetchSize(Integer fetchSize) {
        this.fetchSize = fetchSize;
    }

    public boolean hasNext() {
        return fetchStart + readIndex < writeIndex;
    }

    public T getNext() {
        if (readIndex >= partList.size()) {
            readFromFile();
        }
        return partList.get(readIndex);
    }

    public T next() {
        if (readIndex >= partList.size()) {
            readFromFile();
        }
        return partList.get(readIndex++);
    }

    private void readFromFile() {
        fetchStart = fetchEnd;
        fetchEnd += fetchSize;
        if (fetchEnd >= writeIndex) {
            fetchEnd = writeIndex;
        }
        readIndex = 0;
        partList.clear();
        try {
            int fs = fetchEnd - fetchStart;
            for(int i = 0; i < fs; i++) {
                T object = reader.readObject();
                if (object != null) {
                    partList.add(object);
                } else {
                    break;
                }
//                System.out.println(String.format("fetchStart=%d, i=%d", fetchStart, i));
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (Exception e) {
            throw e;
        }
    }

    public void writeToFile(List<T> list) {
        try {
            for (T t : list) {
                writer.writeObject(t);
            }
            writer.flush();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

class ObjectWriter<T> {
    private File file;
    private FileOutputStream fos;
    private ByteBuf buff;
    private int max = 0; // 对象最大占用字节数

    private Field[] fields;

    // 变长字段
    private int varLen = 0;

    public ObjectWriter(Class<T> clazz, ByteBuf buff, File file) {
        this.buff = buff;
        this.file = file;
        this.fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            Class<?> type = field.getType();
            if(type == String.class) {
                varLen++;
            }
        }
    }

    public void writeObject(T object) throws IOException {
        // 总长度、变长字段长度、null值字段
        // 8      4 * n        n

        // 总长度
        int len = 4;
        ByteBuf totalLen = buff.slice(buff.writeIndex(), 4);
        buff.writerIndex(buff.writeIndex() + 4);

        // 变长字段
        ByteBuf varByteBuf = buff.slice(buff.writeIndex(), varLen * 4);
        buff.writerIndex(buff.writeIndex() + varLen * 4);
        len += varLen * 4;

        // null值字段
        int fl = fields.length;
        int nullRowSize = Bits.NullRowSizeFor(fl);
        RowByte rowByte = new RowByte(buff.slice(buff.writeIndex(), nullRowSize));
        buff.writerIndex(buff.writeIndex() + nullRowSize);
        len += nullRowSize;

        // 数据字段
        for (int i = 0; i < fields.length; i++) {
            Field field = fields[i];

            Object value = null;
            try {
                field.setAccessible(true);
                value = field.get(object);
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }

            if(value == null) {
                rowByte.setNull(i);
                continue;

            } else {
                rowByte.setNotNull(i);
            }

            Class<?> type = field.getType();
            if (type == Integer.class) {
                len += 4;
                buff.writeInt((int)value);

            } else if(type == Long.class){
                len += 8;
                buff.writeLong((Long) value);

            } else if(type == String.class) {
                byte[] bytes = value.toString().getBytes("utf-8");
                // 长度
                varByteBuf.writeInt(bytes.length);
//                System.out.println("字符串长度：" + bytes.length);
                // 内容
                buff.writeBytes(bytes);
                // 增加长度
                len += bytes.length;
            }
        }

        // 总长度
        totalLen.writeInt(len);
        max = len > max ? len : max;

//        System.out.println(object.toString() + "  " + "写入长度:" + len);

        // 写入磁盘
        if(max * 2 > buff.capacity() - buff.writeIndex()) {
            flush();
        }
    }

    public void flush() throws IOException {
        byte[] array = buff.array();
        if (fos == null) {
            try {
                this.fos = new FileOutputStream(file);
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            }
        }
        fos.write(array, buff.readIndex(), buff.writeIndex() - buff.readIndex());
        buff.reset();
    }

}


class ObjectReader<T> {
    private FileInputStream fis;
    private ByteBuf buff;
    private Class<T> clazz;
    private Field[] fields;
    // 变长字段
    private int varLen = 0;

    public ObjectReader(Class<T> clazz, ByteBuf buff, File file) {
        this.clazz = clazz;
        this.buff = buff;
        try {
            this.fis = new FileInputStream(file);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }

        this.fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            Class<?> type = field.getType();
            if(type == String.class) {
                varLen++;
            }
        }
    }

    public T readObject() throws IOException, IllegalAccessException {
        // 总长度、变长字段长度、null值字段
        // 8      8 * n        n
        T object = null;
        try {
            object = clazz.newInstance();
        } catch (InstantiationException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }

        if (buff.writeIndex() - buff.readIndex() < 4) {
            fillBuffer();
            if (buff.writeIndex() - buff.readIndex() == 0) {
                return null;
            }
        }

        // 总长度
        ByteBuf totalLen = buff.slice(buff.readIndex(), 4);
        int len = totalLen.getInt();
        buff.readerIndex(buff.readIndex() + 4);

        if (buff.writeIndex() - buff.readIndex() < len - 4) {
            fillBuffer();
        }

        // 变长字段
        ByteBuf varByteBuf = buff.slice(buff.readIndex(), varLen * 4);
        buff.readerIndex(buff.readIndex() + varLen * 4);

        // null值字段
        int fl = fields.length;
        int nullRowSize = Bits.NullRowSizeFor(fl);
        RowByte rowByte = new RowByte(buff.slice(buff.readIndex(), nullRowSize));
        buff.readerIndex(buff.readIndex() + nullRowSize);

        //
        for (int i = 0; i < fields.length; i++) {
            Field field = fields[i];
            if(rowByte.isNull(i)){
                continue;
            }

            Class<?> type = field.getType();
            Object value = null;
            if (type == Integer.class) {
                value = buff.readInt();

            } else if(type == Long.TYPE){
                value = buff.readLong();

            } else if(type == String.class) {
                int strLen = varByteBuf.readInt();
                value = buff.readString(strLen);
            }

            field.setAccessible(true);
            field.set(object, value);
        }

        return object;
    }

    public void fillBuffer() throws IOException {
        buff.discard();

        byte[] array = buff.array();

        int len = fis.read(array, buff.writeIndex(), buff.capacity() - buff.writeIndex());
        buff.writerIndex(buff.writeIndex() + len);
    }

}

interface ByteBuf {

    int capacity();

    int readIndex();

    int writeIndex();

    void writerIndex(int writerIndex);

    void readerIndex(int readerIndex);

    ByteBuf slice(int index, int length);

    // 写
    void writeInt(Integer value);

    void writeLong(Long value);

    void writeBytes(byte[] value);

    void setByte(int index, byte value);

    // 读
    int readInt();

    long readLong();

    String readString(int len);

    int getInt();

    byte getByte(int index);

    // 其它
    byte[] array();

    void reset();

    void discard();
}

class HeapByteBuf implements ByteBuf {
    private int capacity;
    private int readIndex;
    private int writeIndex;
    private byte[] memory;

    public HeapByteBuf(int capacity){
        this.memory = new byte[capacity];
        this.capacity = capacity;
    }

    @Override
    public int capacity() {
        return capacity;
    }

    @Override
    public int readIndex() {
        return readIndex;
    }

    @Override
    public int writeIndex() {
        return writeIndex;
    }

    @Override
    public void writerIndex(int writerIndex) {
        this.writeIndex = writerIndex;
    }

    @Override
    public void readerIndex(int readerIndex) {
        this.readIndex = readerIndex;
    }

    @Override
    public ByteBuf slice(int index, int length) {
        return new SliceByteBuf(this, index, length);
    }

    @Override
    public void writeInt(Integer value) {
        Bits.setInt(memory, writeIndex, value);
        writeIndex += 4;
    }

    @Override
    public void writeLong(Long value) {
        Bits.setLong(memory, writeIndex, value);
        writeIndex += 8;
    }

    @Override
    public void writeBytes(byte[] value) {
        System.arraycopy(value, 0, memory, writeIndex, value.length);
        writeIndex += value.length;
    }

    @Override
    public void setByte(int index, byte value) {
        memory[index] = value;
    }

    @Override
    public int readInt() {
        checkReadIndex(4);
        int value = Bits.getInt(memory, readIndex);
        readIndex += 4;
        return value;
    }

    private void checkReadIndex(int len){
        if (readIndex + len > writeIndex) {
            throw new ArrayIndexOutOfBoundsException(
                    String.format("访问越界, readIndex=%d, writerIndex=%d, len=%d", readIndex, writeIndex, len)
            );
        }
    }

    @Override
    public long readLong() {
        checkReadIndex(8);
        long value = Bits.getLong(memory, readIndex);
        readIndex += 8;
        return value;
    }

    @Override
    public String readString(int len) {
        checkReadIndex(len);
        String value = new String(memory, readIndex, len);
        readIndex += len;
        return value;
    }

    @Override
    public int getInt() {
        checkReadIndex(4);
        int value = Bits.getInt(memory, readIndex);
        return value;
    }

    @Override
    public byte getByte(int index) {
        checkReadIndex(1);
        return memory[index];
    }

    @Override
    public byte[] array() {
        return memory;
    }

    @Override
    public void reset() {
        readIndex = 0;
        writeIndex = 0;
    }

    @Override
    public void discard() {
        System.arraycopy(memory, readIndex, memory, 0, writeIndex - readIndex);
        writeIndex = writeIndex - readIndex;
        readIndex = 0;
    }

}

class SliceByteBuf implements ByteBuf {
    private int capacity;
    private int readIndex;
    private int writeIndex;

    private int index;
    private ByteBuf buf;

    public SliceByteBuf(ByteBuf buf, int index, int length) {
        this.buf = buf;
        this.index = index;
        this.capacity = length;
    }

    private int idx(Integer idx) {
        return this.index + idx;
    }

    @Override
    public int capacity() {
        return this.capacity;
    }

    @Override
    public int readIndex() {
        return readIndex;
    }

    @Override
    public int writeIndex() {
        return writeIndex;
    }

    @Override
    public void writerIndex(int writerIndex) {
        this.writeIndex = writerIndex;
    }

    @Override
    public void readerIndex(int readerIndex) {
        this.readIndex = readerIndex;
    }

    @Override
    public ByteBuf slice(int index, int length) {
        return new SliceByteBuf(this, index, length);
    }

    @Override
    public void writeInt(Integer value) {
        Bits.setInt(buf.array(), idx(readIndex), value);
        readIndex += 4;
    }

    @Override
    public void writeLong(Long value) {
        Bits.setLong(buf.array(), idx(readIndex), value);
        readIndex += 4;
    }

    @Override
    public void writeBytes(byte[] value) {
        System.arraycopy(value, 0, buf.array(), idx(readIndex), value.length);
        readIndex += value.length;
    }

    @Override
    public void setByte(int index, byte value) {
        buf.array()[idx(index)] = value;
    }

    @Override
    public int readInt() {
        int value = Bits.getInt(buf.array(), idx(readIndex));
        readIndex += 4;
        return value;
    }

    @Override
    public long readLong() {
        long value = Bits.getLong(buf.array(), idx(readIndex));
        readIndex += 8;
        return value;
    }

    @Override
    public String readString(int len) {
        String value = new String(buf.array(), idx(readIndex), len);
        readIndex += len;
        return value;
    }

    @Override
    public int getInt() {
        return Bits.getInt(buf.array(), idx(readIndex));
    }

    @Override
    public byte getByte(int index) {
        return buf.array()[idx(index)];
    }

    @Override
    public byte[] array() {
        return buf.array();
    }

    @Override
    public void reset() {
        readIndex = 0;
        writeIndex = 0;
    }

    @Override
    public void discard() {
        System.arraycopy(buf.array(), idx(readIndex), buf.array(), idx(0), writeIndex - readIndex);
        writeIndex = writeIndex - readIndex;
        readIndex = 0;
    }
}

class RowByte{
    private ByteBuf buf;

    public RowByte(ByteBuf buf){
        this.buf = buf;
    }

    public void setNull(int index) {
        int i = index / 8;
        int pos = index % 8;
        byte b = buf.getByte(i);

        byte pb = (byte)(1 << pos);
        b = (byte)(b | pb);

        buf.setByte(i, b);
    }

    public void setNotNull(int index) {
        int i = index / 8;
        int pos = index % 8;

        byte b = buf.getByte(i);

        byte pb = (byte)(1 << pos);
        b = (byte)(b & pb);

        buf.setByte(i, b);
    }

    public boolean isNull(int index){
        int i = index / 8;
        int pos = index % 8;

        byte b = buf.getByte(i);

        byte pb = (byte)(1 << pos);
        byte c = (byte)(b & pb);

        return b != c;
    }

}

class Bits {
    static final int NullRowSizeFor(int cap) {
        int n = cap - 1;
        n |= n >>> 1;
        n |= n >>> 2;
        n |= n >>> 4;
        n |= n >>> 8;
        n |= n >>> 16;
        n = (n < 0) ? 1 : n + 1;
        n = n / 8;
        return n <= 0 ? 1 : n;
    }

    public static void setInt(byte[] memory, int index, int value) {
        memory[index]     = (byte) (value >>> 24);
        memory[index + 1] = (byte) (value >>> 16);
        memory[index + 2] = (byte) (value >>> 8);
        memory[index + 3] = (byte) value;
    }

    public static void setLong(byte[] memory, int index, long value) {
        memory[index]     = (byte) (value >>> 56);
        memory[index + 1] = (byte) (value >>> 48);
        memory[index + 2] = (byte) (value >>> 40);
        memory[index + 3] = (byte) (value >>> 32);
        memory[index + 4] = (byte) (value >>> 24);
        memory[index + 5] = (byte) (value >>> 16);
        memory[index + 6] = (byte) (value >>> 8);
        memory[index + 7] = (byte) value;
    }

    static int getInt(byte[] memory, int index) {
        return  (memory[index]     & 0xff) << 24 |
                (memory[index + 1] & 0xff) << 16 |
                (memory[index + 2] & 0xff) <<  8 |
                memory[index + 3] & 0xff;
    }

    static long getLong(byte[] memory, int index) {
        return  ((long) memory[index]     & 0xff) << 56 |
                ((long) memory[index + 1] & 0xff) << 48 |
                ((long) memory[index + 2] & 0xff) << 40 |
                ((long) memory[index + 3] & 0xff) << 32 |
                ((long) memory[index + 4] & 0xff) << 24 |
                ((long) memory[index + 5] & 0xff) << 16 |
                ((long) memory[index + 6] & 0xff) <<  8 |
                (long) memory[index + 7] & 0xff;
    }
}